from xtquant import xtdata
import time, os, sys, logging,json
from datetime import datetime
from collections import deque
import pandas as pd
from typing import Dict, Optional
from 基础配置 import settings
from 公共函数 import add_updown, update_updown, update_stock_data,get_stock
from 组件.notice import show_notification_thread
from 组件.rabbitmq import RabbitMQTickPublisher
from 指标.MACD import MACDSignalCalculator,MACDConfig, CacheConfig
from 指标.日内九转 import THSNineTurn
from 指标.五分钟日内九转 import THSNineTurn5m
from 指标.波动率 import ATRCalculator
from 指标.RSI import IndependentRSICalculator
from 指标.VMA import IndependentVMACalculator
from 指标.KDJ import IndependentKDJCalculator
from 指标.涨跌幅 import UpDownCalculator
from 指标.交易总量 import get_stock_trading_volume
from 指标.MA import IndependentMACalculator
from 异步指标计算器 import AsyncIndicatorCalculator
from 数据预加载器 import DataPreloader
from 性能监控器 import PerformanceMonitor

class MarketMonitor:
    ##初始化
    def __init__(self, trader=None, log_callback=None, trade_cmd_queue=None):
        # 配置常量
        self.current_date = datetime.now().strftime("%Y-%m-%d")
        self.VOLUME_SCALE_THRESHOLD = 100000  # 成交量缩放阈值
        self.VOLUME_SCALE_FACTOR = 100  # 成交量缩放因子
        self.MACD_DOWNLOAD_DELAY = 0.3  # MACD数据下载延迟（秒）
        self.NINE_TURN_DOWNLOAD_DELAY = 1.0  # 九转数据下载延迟（秒）
        self.day_look = settings.DAY_LOOK

        # 交易时间段常量
        self.MORNING_START = datetime.strptime("09:30", "%H:%M").time()
        self.MORNING_END = datetime.strptime("11:30", "%H:%M").time()
        self.AFTERNOON_START = datetime.strptime("13:00", "%H:%M").time()
        self.AFTERNOON_END = datetime.strptime("15:00", "%H:%M").time()

        self.stock_codes = [s["all_code"] for s in settings.STOCK_CODE]
        self.period = "tick"
        self.stock_type = {stock["all_code"]: stock["name"] for stock in settings.STOCK_CODE}
        self.stock_min_rsi = {stock["all_code"]: stock["min_rsi"] for stock in settings.STOCK_CODE}

        self.historical_data = {code: deque(maxlen=100) for code in self.stock_codes}
        self.fluctuation_status = {code: {"in_range": False, "start_time": "", "updown_id": 0} for code in
                                   self.stock_codes}
        self.log_callback = log_callback
        self._running = False
        self.trader = trader
        self.trade_cmd_queue = trade_cmd_queue
        self.xtdata = xtdata

        # MACD相关初始化
        config = MACDConfig.get_optimized_params()
        self.macd_calculator = MACDSignalCalculator(self.xtdata, use_warmup=config['use_warmup'])
        self.macd_fast = MACDConfig.FAST_EMA
        self.macd_slow = MACDConfig.SLOW_EMA
        self.macd_signal = MACDConfig.SIGNAL_EMA
        self.tick_counters = {code: 0 for code in self.stock_codes}  # 每个股票的tick计数器
        self.macd_cache = {}  # MACD结果缓存，避免重复计算
        self.last_macd_check = {code: 0 for code in self.stock_codes}  # 上次MACD检查时间
        self.macd_check_interval = MACDConfig.TICK_CHECK_INTERVAL  # 每N个tick检查一次MACD
        self.macd_cache_ttl = config['cache_ttl']  # MACD缓存有效期（秒）
        self.max_cache_size = config['max_cache_size']  # 最大缓存大小
        self.macd_values_cache = {}  # MACD数值缓存，存储DIFF和DEA值

        # 添加信号状态跟踪，避免重复提示和下单
        self.signal_status = {code: {
            "last_golden_cross": None, 
            "last_death_cross": None,
            "last_golden_time": 0,
            "last_death_time": 0
        } for code in self.stock_codes}
        # 维护价格历史用于EMA计算（每只股票最近100个tick）
        self.price_history = {code: deque(maxlen=100) for code in self.stock_codes}
        # 日内九转（用于 513090.SH），采用同花顺风格实现
        self.nine_turn = THSNineTurn(compare_offset=4,required_bars=10,verbose=True,enable_ma_filter=False, ma_period=20,  use_5min_ma=True)
        self.nine_turn5m = THSNineTurn5m(compare_offset=4, required_bars=9, verbose=True, enable_ma_filter=False,
                                     ma_period=20, use_5min_ma=True)
        # ATR 波动率计算器（用于金叉买入时的波动过滤）
        self.atr_calculator = ATRCalculator(xtdata, lookback=settings.ATR_LOOKBACK, method='ema', cache_ttl_seconds=10,
                                            verbose=False)
        # RSI 计算器（严格同花顺 RSI10，仅在收盘确认时更新），B方案：tick分钟聚合+预览RSI
        # RSI 计算器（完全独立，直接获取历史数据计算，无需预热）
        self.rsi_calculator = IndependentRSICalculator(period=5, xtdata=xtdata)
        self._minute_bucket: Dict[str, str] = {}
        self._minute_close: Dict[str, float] = {}
        # 添加RSI超卖买入信号时间保护
        self.rsi_buy_protection_time = {}  # 记录每只股票最后一次RSI买入信号的时间
        self.rsi_buy_protection_minutes = 10  # RSI买入信号保护时间（分钟）
        # 添加死叉保护时间，防止死叉后立即买入
        self.death_cross_protection_time = {}  # 记录每只股票最后一次死叉信号的时间
        self.death_cross_protection_minutes = settings.DEATH_CROSS_PROTECTION_MINUTES  # 死叉保护时间（分钟）
        # VMA成交量指标（使用独立的VMA计算器）
        self.vma_calculator = IndependentVMACalculator(period=30, factor=1.2, xtdata=xtdata)
        # KDJ指标计算器
        self.kdj_calculator = IndependentKDJCalculator(k_period=9, d_period=3, xtdata=xtdata, use_warmup=True)
        # 涨跌幅指标器
        self.updown_calculator = UpDownCalculator(xtdata=xtdata)
        # MA指标计算器
        self.ma_calculator = IndependentMACalculator(xtdata=xtdata)
        # 存储每只股票上一笔的"累计成交量"以换算为每tick增量
        self._last_cum_volume: Dict[str, float] = {}
        # 是否在启动时预热 RSI 与 成交量（下载当日tick并聚合分钟）
        self.enable_warmup: bool = True
        self.saver = RabbitMQTickPublisher()
        
        # 价格跌幅检查阈值
        self.price_drop_threshold = 0.02  # 2%跌幅阈值
        
        # 波动率检查相关配置
        self.volatility_check_period = 10  # 检查前10分钟的波动率
        self.volatility_threshold = 0.003  # 0.3%波动率阈值
        self.price_history_10min = {code: deque(maxlen=600) for code in self.stock_codes}  # 存储10分钟内的价格历史

        # 初始化异步指标计算器
        self.async_calculator = AsyncIndicatorCalculator(self)
        
        # 初始化数据预加载器
        self.data_preloader = DataPreloader(self.stock_codes, self.xtdata)
        
        # 初始化性能监控器
        self.performance_monitor = PerformanceMonitor()

        # 记录系统启动时间，用于5分钟内不买入的保护机制
        self.system_start_time = time.time()
        
        # 初始化日志系统
        self._setup_logging()
    ##日志系统
    def _setup_logging(self):
        """初始化简化的日志系统"""
        try:
            # 创建logs目录
            logs_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'logs')
            if not os.path.exists(logs_dir):
                os.makedirs(logs_dir, exist_ok=True)

            # 设置信号日志文件 - 使用日期格式
            log_file = os.path.join(logs_dir, f'signals_{self.current_date}.log')

            # 创建logger
            self.signal_logger = logging.getLogger("signals")
            self.signal_logger.setLevel(logging.INFO)
            self.signal_logger.propagate = False

            # 清除已有的handlers
            for handler in self.signal_logger.handlers[:]:
                self.signal_logger.removeHandler(handler)

            # 添加文件handler - 使用简单的FileHandler，因为文件名已经包含日期
            file_handler = logging.FileHandler(
                filename=log_file,
                encoding="utf-8"
            )
            formatter = logging.Formatter(
                "%(asctime)s - %(levelname)s - %(message)s",
                datefmt="%Y-%m-%d %H:%M:%S"
            )
            file_handler.setFormatter(formatter)
            self.signal_logger.addHandler(file_handler)

        except Exception as e:
            # 日志系统初始化失败时，直接打印错误信息
            print(f"⚠️ 日志系统初始化失败: {e}")
            self.signal_logger = None
    def _check_date_change(self):
        """检查日期是否变化，如果变化则重新初始化日志系统"""
        today = datetime.now().strftime("%Y-%m-%d")
        if today != self.current_date:
            self.current_date = today
            self._setup_logging()
    def log(self, msg):
        """统一的日志记录方法：同时输出到控制台和文件"""
        # 检查日期变化
        self._check_date_change()
        
        print(msg)
        if self.log_callback:
            self.log_callback(msg)

        # 同时记录到信号日志文件
        if hasattr(self, 'signal_logger') and self.signal_logger:
            try:
                self.signal_logger.info(msg)
            except Exception:
                pass  # 如果文件日志失败，不影响控制台输出
    def log_signal(self, msg: str, level: str = "info"):
        """记录交易信号到日志文件"""
        # 检查日期变化
        self._check_date_change()
        
        if hasattr(self, 'signal_logger') and self.signal_logger:
            try:
                if level == "warning":
                    self.signal_logger.warning(msg)
                elif level == "error":
                    self.signal_logger.error(msg)
                else:
                    self.signal_logger.info(msg)
            except Exception:
                pass
        # 同时输出到控制台
        print(msg)

    ##订阅tick回调
    def quote_callback(self, data):
        # 记录处理开始时间
        start_time = time.time()
        
        for stock_code, stock_data in data.items():
            stock_name = self.stock_type.get(stock_code)
            # 取最新一笔：time 最大
            try:
                if isinstance(stock_data, list) and stock_data:
                    def _as_ts(x):
                        """将时间戳转换为可比较的数值，返回None表示无效"""
                        try:
                            t = x.get('time') if isinstance(x, dict) else None
                            if t is None:
                                return None
                            if isinstance(t, (int, float)):
                                return float(t)
                            s = str(t)
                            if s.isdigit():
                                return float(s)
                            ts = pd.to_datetime(s, errors='coerce')
                            if pd.isna(ts):
                                return None
                            return float(ts.timestamp() * 1000.0)
                        except Exception:
                            return None

                    # 过滤掉无效时间戳的数据，然后选择最新的
                    valid_data = [(item, ts) for item in stock_data for ts in [_as_ts(item)] if ts is not None]
                    if valid_data:
                        data0 = max(valid_data, key=lambda x: x[1])[0]
                    else:
                        # 如果没有有效时间戳，取最后一个元素
                        data0 = stock_data[-1]
                        self.log(f"⚠️ {stock_name} 所有数据时间戳无效，使用最后一条数据")
                else:
                    data0 = stock_data[0]
            except Exception as e:
                self.log(f"⚠️ {stock_name} 数据处理异常: {e}")
                data0 = stock_data[0] if isinstance(stock_data, list) else stock_data
                
            # 先获取时间戳进行过期检查
            timestamp = data0.get('time')
            
            # 过滤过期数据（超过1分钟的数据）- 放在最前面
            if timestamp is not None:
                try:
                    if isinstance(timestamp, (int, float)):
                        if timestamp > 1e12:  # 毫秒时间戳
                            tick_datetime = datetime.fromtimestamp(timestamp / 1000)
                        else:  # 秒时间戳
                            tick_datetime = datetime.fromtimestamp(timestamp)
                        
                        time_diff = (datetime.now() - tick_datetime).total_seconds()
                        if time_diff > 60:  # 超过1分钟的数据
                            self.log(f"⚠️ {stock_name} 数据过期: 延迟{time_diff:.0f}秒，跳过处理")
                            continue
                except Exception as e:
                    self.log(f"⚠️ {stock_name} 时间戳解析失败: {e}")
                    # 如果时间戳解析失败，仍然处理数据
            
            # 获取其他数据
            last_price = data0.get('lastPrice')
            volume = data0.get('volume', 0)
            
            # 简化日志输出，只在每100个tick输出一次
            if self.tick_counters[stock_code] % 100 == 0:
                current_time = datetime.now().strftime("%H:%M:%S")
                print(f"tick监控-{stock_name}-{last_price} (当前:{current_time})")

            # 数据验证
            if last_price is None or timestamp is None:
                self.log(f"⚠️ {stock_name} 数据无效: price={last_price}, timestamp={timestamp}")
                continue

            # 增加tick计数器
            self.tick_counters[stock_code] += 1
            # 记录价格到历史用于EMA计算
            self.price_history[stock_code].append(last_price)
            
            # 记录价格到10分钟历史用于波动率计算
            current_time = time.time()
            self.price_history_10min[stock_code].append({
                'price': last_price,
                'timestamp': current_time
            })

            ##获得指标 - 使用异步计算优化性能
            macd = ""
            updown_id = 0
            diefu = 0
            
            # 分钟聚合与落库：在分钟切换时先用上一分钟最后价update，当前分钟用预览RSI显示
            try:
                minute_key = self._to_minute_bucket(timestamp)
                if minute_key is not None:
                    prev_key = self._minute_bucket.get(stock_code)
                    if prev_key is None:
                        self._minute_bucket[stock_code] = minute_key
                        self._minute_close[stock_code] = float(last_price)
                    elif prev_key == minute_key:
                        self._minute_close[stock_code] = float(last_price)
                    else:
                        prev_close = self._minute_close.get(stock_code)
                        # 仅在交易时段内的上一分钟才进行确认更新
                        if prev_close is not None and self._is_session_minute(prev_key):
                            pass
                        self._minute_bucket[stock_code] = minute_key
                        self._minute_close[stock_code] = float(last_price)
            except Exception:
                pass

            # 提交异步计算任务
            self.async_calculator.submit_calculation(stock_code, 'updown', last_price, timestamp)
            self.async_calculator.submit_calculation(stock_code, 'macd', last_price, timestamp)
            # self.async_calculator.submit_calculation(stock_code, 'rsi', last_price, timestamp)
            # self.async_calculator.submit_calculation(stock_code, 'nine_turn', last_price, timestamp)
            ##self.async_calculator.submit_calculation(stock_code, 'nine_turn5m', last_price, timestamp)

            # 获取缓存的指标结果
            cached_results = self.async_calculator.get_all_cached_results(stock_code)

            # 2. RSI指标
            rsi = 0
            rsi_confirm = 0
            
            # 4. 涨跌幅指标
            updown_info = cached_results.get('updown')
            if updown_info:
                diefu = updown_info.get("updown", 0)
            else:
                updown_info = None

            # 这里可以添加针对不同股票的处理逻辑
            if stock_code == "1513130.SH" or stock_code == "1513060.SH" or  stock_code == "1159509.SZ":
                # 恒生科技和医疗使用波动检测
                # updown_id = self.updown_watch_tick(stock_code, stock_data)
                pass
            else:
                # 获取MACD信号
                macd = cached_results.get('macd', "")

                # 获取九转信号
                nine_turn_signal = cached_results.get('nine_turn')
                if nine_turn_signal == "买入":
                    macd = "九转买"
                elif nine_turn_signal == "卖出":
                    macd = "九转卖"

            # 触发信号时偏向用"已确认RSI"，其余时段用"预览RSI"，与回测口径一致
            if macd in ("金叉", "死叉", "九转买", "九转卖"):
                rsi_for_decision = rsi_confirm if rsi_confirm is not None else rsi
            else:
                rsi_for_decision = rsi if rsi is not None else rsi_confirm
                
            try:
                update_stock_data(stock_code, now_price=last_price, macd=macd, updown_id=updown_id,rsi=rsi_for_decision,diefu=diefu)
            except Exception as e:
                print(f"数据更新错误")

            self.check_buy_sell(stock_code, stock_name, last_price, macd, rsi_for_decision, updown_info, timestamp)
        
        # 记录处理时间并更新性能监控
        processing_time = time.time() - start_time
        self.performance_monitor.record_tick_processing(processing_time)
        
        # 记录队列大小和缓存命中率
        queue_size = self.async_calculator.get_queue_size()
        self.performance_monitor.record_queue_size(queue_size)
        
        cache_stats = self.async_calculator.get_cache_stats()
        if 'hit_rate' in cache_stats:
            hit_rate = float(cache_stats['hit_rate'].replace('%', ''))
            self.performance_monitor.record_cache_hit_rate(hit_rate)
    ##这个方法是专门计算是否需要买入或者卖出
    def check_buy_sell(self, stock_code: str, stock_name: str, last_price: float, macd: str, rsi_value: Optional[float], updown_info: Dict = None, timestamp = None):
        need_buy = False
        need_sell = False
        current_time = datetime.now().strftime("%H:%M:%S")

        atr_ratio = None
        threshold = None
        up_0 = False
        if macd == "金叉":
            self.log_signal(f"{stock_name}:金叉信号")

            if updown_info:
                if updown_info.get("updown", 0) <= -5:
                    if stock_code == "513130.SH" or stock_code == "513060.SH" or stock_code == "159509.SZ":
                        pass
                    else:
                        if macd:
                            self.log_signal(f"🚫 {stock_name}{macd}但是 交易被暂停: 跌幅{updown_info.get("updown", 0)}")
                        return

                if updown_info.get("updown", 0) >= 3:
                    if macd:
                        self.log_signal(f"🚫 {stock_name}{macd}但是 交易被暂停: 涨幅{updown_info.get("updown", 0)}")
                    return
            else:
                self.log_signal(f"🚫 {stock_name}交易被暂停: 未检测到涨跌幅")
                return

            if stock_code == "513130.SH" or stock_code == "513060.SH" or  stock_code == "159509.SZ":
                need_buy = True
                self.log_signal(f"{stock_name} 出现{macd}直接买入")

            # # 🔥 同步更新5分钟九转状态，确保获取最新的计数值
            # try:
            #     # 使用传入的timestamp，如果没有则使用当前时间
            #     ts = timestamp if timestamp is not None else time.time()
            #     self.nine_turn5m.on_tick(stock_code, ts, last_price)
            # except Exception as e:
            #     self.log_signal(f"⚠️ {stock_name} 5分钟九转状态更新失败: {e}")
            #
            # # 现在获取的 down_count 是最新的
            # current_down_count = self.nine_turn5m.get_down_count(stock_code)
            # if current_down_count >= 5:
            #     self.log_signal(f"不符合:5分中线连续下跌次数{current_down_count}")
            #     return
            # else:
            #     self.log_signal(f"{stock_code} 5分中线连续下跌次数{current_down_count}")

            can_buy = 0
            # 从缓存中获取MACD数值来判断是0上金叉还是0下金叉
            try:
                macd_values = self.macd_values_cache.get(stock_code, {})
                diff = macd_values.get('diff', 0)
                dea = macd_values.get('dea', 0)
                # 判断是0上金叉还是0下金叉
                if diff > 0 and dea > 0:
                    golden_cross_type = "0上金叉"
                else:
                    golden_cross_type = "0下金叉"
            except Exception as e:
                self.log_signal(f"获取MACD数值失败: {e}")
                golden_cross_type = "金叉"  # 默认值

            need_buy = True
            self.log_signal(f"金叉买入触发：{stock_name}")

        if macd == "死叉" or macd == "九转卖":
            need_sell = True
            self.log_signal(f"卖出触发：{stock_code} macd={macd}")

        if macd == "九转买":
            pass

        order_remark = {
            "level":0,
            "macd": macd,
        }
        order_remark_string = json.dumps(order_remark, ensure_ascii=False)

        if need_sell:
            # self.execute_death_cross_sell(stock_code, stock_name,macd)
            pass
        if need_buy:
            self.execute_golden_cross_buy(stock_code, stock_name, last_price,order_remark_string)

    ##检查MACD金叉死叉信号
    def check_macd_signals(self, stock_code: str, current_price: float):
        current_time = time.time()
        stock_name = self.stock_type.get(stock_code)

        # 检查缓存是否有效
        if (CacheConfig.should_use_cache() and
                stock_code in self.macd_cache and
                MACDConfig.is_cache_valid(self.last_macd_check[stock_code], current_time)):
            # 使用缓存的结果
            cached_result = self.macd_cache[stock_code]
            macd = self.process_macd_signals(stock_code, stock_name, current_price, cached_result)
            return macd

        # 检查缓存大小，如果超过限制则清理
        if len(self.macd_cache) > self.max_cache_size:
            self._cleanup_cache()

        try:
            # 异步计算MACD，避免阻塞主线程
            if MACDConfig.ENABLE_DETAILED_LOGGING:
                self.log(f"🔄 {stock_name} 开始计算MACD信号...")

            # 获取MACD信号（使用同花顺方法：包含集合竞价，简单初始化）
            macd_signals = self.macd_calculator.detect_macd_signals(
                stock_code,
                fast=self.macd_fast,
                slow=self.macd_slow,
                signal=self.macd_signal,
                include_auction=True,      # ⭐ 包含集合竞价（9:15开始）
                use_simple_init=True       # ⭐ 使用同花顺方法（第一个EMA=第一个收盘价）
            )

            # 缓存结果
            self.macd_cache[stock_code] = macd_signals
            self.last_macd_check[stock_code] = current_time
            
            # 缓存MACD数值（DIFF和DEA）
            latest_macd = macd_signals.get('latest_macd', {})
            if latest_macd:
                self.macd_values_cache[stock_code] = {
                    'diff': latest_macd.get('diff', 0),
                    'dea': latest_macd.get('dea', 0),
                    'timestamp': current_time
                }

            # 处理MACD信号
            macd = self.process_macd_signals(stock_code, stock_name, current_price, macd_signals)

            return macd

        except Exception as e:
            self.log(f"❌ {stock_name} MACD计算异常: {e}")
            import traceback
            traceback.print_exc()
    ##清理缓存
    def _cleanup_cache(self):
        current_time = time.time()
        expired_keys = []

        for code, last_check in self.last_macd_check.items():
            if current_time - last_check > self.macd_cache_ttl:
                expired_keys.append(code)

        # 删除过期缓存
        for code in expired_keys:
            if code in self.macd_cache:
                del self.macd_cache[code]
            if code in self.last_macd_check:
                del self.last_macd_check[code]
            if code in self.macd_values_cache:
                del self.macd_values_cache[code]

        # 如果缓存仍然过大，删除最旧的条目
        if len(self.macd_cache) > self.max_cache_size:
            sorted_codes = sorted(self.last_macd_check.items(), key=lambda x: x[1])
            codes_to_remove = len(self.macd_cache) - self.max_cache_size

            for i in range(codes_to_remove):
                if i < len(sorted_codes):
                    code = sorted_codes[i][0]
                    if code in self.macd_cache:
                        del self.macd_cache[code]
                    if code in self.last_macd_check:
                        del self.last_macd_check[code]
                    if code in self.macd_values_cache:
                        del self.macd_values_cache[code]
    ##MACD信号预热
    def _warmup_macd_signals(self):
        """在系统启动时预热MACD信号状态，避免重启后重复触发历史信号"""
        try:
            self.log("🔥 开始MACD信号预热...")
            today = datetime.now().strftime("%Y%m%d")
            
            for stock_code in self.stock_codes:
                try:
                    stock_name = self.stock_type.get(stock_code, stock_code)
                    
                    # 下载并获取当天的tick数据
                    try:
                        xtdata.download_history_data(stock_code=stock_code, period="tick", start_time=today, end_time=today)
                        time.sleep(self.MACD_DOWNLOAD_DELAY)
                    except Exception as e:
                        self.log(f"⚠️ {stock_name} 下载历史数据失败: {e}")
                        continue
                    
                    # 获取MACD信号（包括历史上的所有金叉和死叉）
                    macd_signals = self.macd_calculator.detect_macd_signals(
                        stock_code,
                        fast=self.macd_fast,
                        slow=self.macd_slow,
                        signal=self.macd_signal,
                        include_auction=True,
                        use_simple_init=True
                    )
                    
                    if not macd_signals:
                        continue
                    
                    # 初始化当前时间戳
                    current_timestamp = time.time()
                    
                    # 处理金叉信号
                    if macd_signals.get('golden_cross') and len(macd_signals['golden_cross']) > 0:
                        latest_golden = macd_signals['golden_cross'][-1]  # 获取最后一个金叉
                        signal_time = latest_golden.get('time', '')
                        signal_price = latest_golden.get('price', 0)
                        
                        if signal_time:
                            # 记录到signal_status中，防止重复触发
                            signal_key = f"{signal_time}_{signal_price}"
                            self.signal_status[stock_code]["last_golden_cross"] = signal_key
                            self.signal_status[stock_code]["last_golden_time"] = current_timestamp
                            self.log(f"✅ {stock_name} 加载历史金叉信号: {signal_time} @ {signal_price:.3f}")
                    
                    # 处理死叉信号
                    if macd_signals.get('death_cross') and len(macd_signals['death_cross']) > 0:
                        latest_death = macd_signals['death_cross'][-1]  # 获取最后一个死叉
                        signal_time = latest_death.get('time', '')
                        signal_price = latest_death.get('price', 0)
                        
                        if signal_time:
                            # 记录到signal_status中，防止重复触发
                            signal_key = f"{signal_time}_{signal_price}"
                            self.signal_status[stock_code]["last_death_cross"] = signal_key
                            
                            # 解析时间戳并记录死叉保护时间
                            try:
                                signal_datetime = datetime.strptime(signal_time, '%Y-%m-%d %H:%M:%S')
                                signal_timestamp = signal_datetime.timestamp()
                                self.death_cross_protection_time[stock_code] = signal_timestamp
                                self.log(f"✅ {stock_name} 加载历史死叉信号: {signal_time} @ {signal_price:.3f}")
                            except Exception as e:
                                self.log(f"⚠️ {stock_name} 解析死叉时间失败: {e}")
                    
                    # 缓存MACD数值
                    latest_macd = macd_signals.get('latest_macd', {})
                    if latest_macd:
                        self.macd_values_cache[stock_code] = {
                            'diff': latest_macd.get('diff', 0),
                            'dea': latest_macd.get('dea', 0),
                            'timestamp': current_timestamp
                        }
                    
                except Exception as e:
                    self.log(f"⚠️ {stock_name} MACD信号预热失败: {e}")
                    continue
            
            self.log("✅ MACD信号预热完成")
            
        except Exception as e:
            self.log(f"❌ MACD信号预热异常: {e}")
            import traceback
            traceback.print_exc()
    ##重置信号状态，清除历史记录
    def reset_signal_status(self, stock_code: str = None):
        if stock_code:
            # 重置指定股票的信号状态
            if stock_code in self.signal_status:
                self.signal_status[stock_code] = {
                    "last_golden_cross": None, 
                    "last_death_cross": None,
                    "last_golden_time": 0,
                    "last_death_time": 0
                }
            # 重置死叉保护时间
            if stock_code in self.death_cross_protection_time:
                self.death_cross_protection_time[stock_code] = 0
        else:
            # 重置所有股票的信号状态
            for code in self.stock_codes:
                self.signal_status[code] = {
                    "last_golden_cross": None, 
                    "last_death_cross": None,
                    "last_golden_time": 0,
                    "last_death_time": 0
                }
            # 重置所有股票的死叉保护时间
            self.death_cross_protection_time.clear()
    ##获取信号状态信息
    def get_signal_status(self, stock_code: str = None):
        if stock_code:
            return self.signal_status.get(stock_code, {})
        else:
            return self.signal_status
    ##获取死叉保护状态信息
    def get_death_cross_protection_status(self, stock_code: str = None):
        """获取死叉保护状态信息"""
        if stock_code:
            if stock_code in self.death_cross_protection_time:
                last_time = self.death_cross_protection_time[stock_code]
                if last_time > 0:
                    current_time = time.time()
                    time_since_death = (current_time - last_time) / 60
                    remaining_time = max(0, self.death_cross_protection_minutes - time_since_death)
                    return {
                        "protected": remaining_time > 0,
                        "last_death_cross_time": last_time,
                        "time_since_death_minutes": time_since_death,
                        "remaining_protection_minutes": remaining_time
                    }
            return {"protected": False, "last_death_cross_time": 0, "time_since_death_minutes": 0, "remaining_protection_minutes": 0}
        else:
            # 返回所有股票的死叉保护状态
            result = {}
            for code in self.stock_codes:
                result[code] = self.get_death_cross_protection_status(code)
            return result
    ##执行金叉自动买入操作
    def execute_golden_cross_buy(self, stock_code: str, stock_name: str, current_price: float,order_remark:str) -> bool:
        try:
            # 检查是否在交易时间段内
            if not self._is_trading_time():
                self.log_signal(f"交易提示:{stock_name}不在交易时间内")
                return False

            code_to_volume = {s["all_code"]: s["volume"] for s in settings.STOCK_CODE}
            volume = int(code_to_volume.get(stock_code, 100))

            # 检测数据库是否手动关闭单只股票
            stock_data = get_stock(stock_code)
            if stock_data["run"] == 0:
                self.log_signal(f"交易提示:{stock_name}已经手动关闭")
                return False

            # 执行金叉自动买入
            try:
                ##队列交易
                order_data = {
                    "type": "buy",
                    "code": stock_code,
                    "volume": volume,
                    "order_remark": order_remark,
                    "current_price":current_price
                }
                try:
                    self.saver.publish_order(order_data=order_data)
                except Exception as e:
                    # self.log_signal(f"❌{stock_name}队列发送失败: {e}")
                    return False
                # 优先通过指令队列通知主线程执行买入（避免子进程直接触控交易对象）
                if self.trade_cmd_queue:
                    try:
                        trade_cmd = {
                            "action": "buy",
                            "code": stock_code,
                            "volume":volume,
                            "reason": order_remark,
                            "current_price":current_price
                        }
                        self.trade_cmd_queue.put(trade_cmd)
                        return True
                    except Exception as e:
                        self.log_signal(f"❌ {stock_name}发送交易指令到队列失败: {e}")
                        return False
                else:
                    return False
            except Exception as e:
                self.log_signal(f"❌ {stock_name}执行买入异常2: {e}")
                return False
        except Exception as e:
            self.log_signal(f"❌ {stock_name}执行买入异常3: {e}")
            return False

    ##执行死叉自动平仓操作【这里需要模仿卖出操作,调用主线程运行界面的 按照股票代码平仓方法】
    def execute_death_cross_sell(self, stock_code: str,stock_name: str,macd:str) -> bool:
        try:
            code_to_volume = {s["all_code"]: s["volume"] for s in settings.STOCK_CODE}
            volume = int(code_to_volume.get(stock_code, 100))

            # 执行死叉卖出
            try:
                ##队列交易
                order_data = {
                    "type": "sell",
                    "code": stock_code,
                    "volume": volume,
                    "reason": macd,
                }
                try:
                    self.saver.publish_order(order_data=order_data)
                except Exception as e:
                    # self.log_signal(f"❌{stock_name}队列发送失败: {e}")
                    return False
                # 优先通过指令队列通知主线程执行买入（避免子进程直接触控交易对象）
                if self.trade_cmd_queue:
                    try:
                        trade_cmd = {
                            "action": "sell",
                            "code": stock_code,
                            "volume":volume,
                            "reason": macd,
                        }
                        self.trade_cmd_queue.put(trade_cmd)
                        return True
                    except Exception as e:
                        self.log_signal(f"❌ {stock_name}发送交易指令到队列失败: {e}")
                        return False
                else:
                    return False
            except Exception as e:
                self.log_signal(f"❌ {stock_name}执行卖出异常2: {e}")
                return False
        except Exception as e:
            self.log_signal(f"❌ {stock_name}执行卖出异常3: {e}")
            return False
    ##检查当前是否在交易时间段内
    def _is_trading_time(self) -> bool:
        """检查当前是否在交易时间段内"""
        try:
            now = datetime.now()
            current_time = now.time()

            MORNING_START = datetime.strptime("09:35", "%H:%M").time()
            MORNING_END = datetime.strptime("11:20", "%H:%M").time()
            AFTERNOON_START = datetime.strptime("13:05", "%H:%M").time()
            AFTERNOON_END = datetime.strptime("14:40", "%H:%M").time()

            # 检查是否在交易时间段内
            is_morning_trading = MORNING_START <= current_time <= MORNING_END
            is_afternoon_trading = AFTERNOON_START <= current_time <= AFTERNOON_END

            return is_morning_trading or is_afternoon_trading

        except Exception as e:
            self.log(f"❌ 检查交易时间异常: {e}")
            return False
    ##处理MACD信号结果
    def process_macd_signals(self, stock_code: str, stock_name: str, current_price: float, macd_signals: Dict):
        try:
            # 检查金叉信号
            if macd_signals.get('golden_cross'):
                latest_golden = macd_signals['golden_cross'][-1]  # 最新的金叉信号
                signal_time = latest_golden.get('time', '')
                signal_price = latest_golden.get('price', 0)

                # 检查是否是新产生的金叉信号（时间较近）
                if signal_time:
                    try:
                        signal_datetime = datetime.strptime(signal_time, '%Y-%m-%d %H:%M:%S')
                        time_diff = abs((datetime.now() - signal_datetime).total_seconds())

                        # 如果金叉信号在最近指定时间内产生，则发出通知
                        signal_window = MACDConfig.get_signal_window()
                        if time_diff < signal_window:
                            # 先检查时间间隔保护：距离上次金叉时间必须超过60秒
                            current_time = time.time()
                            last_golden_time = self.signal_status[stock_code].get("last_golden_time", 0)
                            time_since_last_golden = current_time - last_golden_time
                            
                            if time_since_last_golden < 300:  # 60秒保护期
                                # 在保护期内，即使信号key不同也不触发
                                return None
                            
                            # 再检查信号key是否相同
                            current_signal_key = f"{signal_time}_{signal_price}"
                            last_signal = self.signal_status[stock_code]["last_golden_cross"]

                            if current_signal_key != last_signal:
                                # 这是新的金叉信号，执行操作
                                self.signal_status[stock_code]["last_golden_cross"] = current_signal_key
                                self.signal_status[stock_code]["last_golden_time"] = current_time
                                self.log_signal(f"{stock_name}:金叉")
                                # 调用封装的买入方法
                                return "金叉"
                            else:
                                # 这是重复的金叉信号，不执行操作
                                return None
                    except Exception as e:
                        self.log(f"❌ 处理金叉信号时间异常: {e}")
                        pass

            # 检查死叉信号
            if macd_signals.get('death_cross'):
                latest_death = macd_signals['death_cross'][-1]  # 最新的死叉信号
                signal_time = latest_death.get('time', '')
                signal_price = latest_death.get('price', 0)

                # 检查是否是新产生的死叉信号（时间较近）
                if signal_time:
                    try:
                        signal_datetime = datetime.strptime(signal_time, '%Y-%m-%d %H:%M:%S')
                        time_diff = abs((datetime.now() - signal_datetime).total_seconds())

                        # 如果死叉信号在最近指定时间内产生，则发出通知
                        signal_window = MACDConfig.get_signal_window()
                        if time_diff < signal_window:
                            # 检查是否与上次死叉信号相同，避免重复提示
                            current_signal_key = f"{signal_time}_{signal_price}"
                            last_signal = self.signal_status[stock_code]["last_death_cross"]

                            if current_signal_key != last_signal:
                                # 这是新的死叉信号，执行操作
                                self.signal_status[stock_code]["last_death_cross"] = current_signal_key
                                # 记录死叉时间，用于保护期计算
                                self.death_cross_protection_time[stock_code] = time.time()

                                msg = f"{stock_name}:死叉"
                                self.log_signal(msg)
                                ##show_notification_thread("MACD死叉", msg)
                                # 调用封装的按代码平仓
                                return "死叉"
                            else:
                                # 这是重复的死叉信号，不执行操作
                                return None
                    except Exception:
                        pass

            # 输出最新MACD状态（减少输出频率，只在有变化时输出）
            latest_macd = macd_signals.get('latest_macd', {})
            if latest_macd:
                diff = latest_macd.get('diff', 0)
                dea = latest_macd.get('dea', 0)
                trend = latest_macd.get('trend', 'unknown')

                # 只在趋势变化时输出
                if stock_code not in self.macd_cache.get('last_trend', {}):
                    self.macd_cache['last_trend'] = {}

                if self.macd_cache['last_trend'].get(stock_code) != trend:
                    self.macd_cache['last_trend'][stock_code] = trend
                    self.log_signal(f"📊 {stock_name} MACD趋势: {trend}, DIFF: {diff:.6f}, DEA: {dea:.6f}")

        except Exception as e:
            self.log(f"❌ {stock_name} 处理MACD信号异常: {e}")
    ##波动检测
    def updown_watch_tick(self, stock_code, stock_data):
        current_time = time.time()
        now = datetime.fromtimestamp(current_time).strftime("%H:%M:%S")
        stock_name = self.stock_type.get(stock_code)
        current_price = stock_data[0].get("lastPrice")  # 根据实际API调整字段名

        # 1. 保存当前数据
        self.historical_data[stock_code].append({
            "price": current_price,
            "time": current_time
        })

        # 2. 检查3分钟内波动（需至少有2个数据点）
        if len(self.historical_data[stock_code]) >= settings.UPDOWN_TIME:
            prices = [item["price"] for item in self.historical_data[stock_code]]
            min_p, max_p = min(prices), max(prices)
            if min_p > 0:
                fluctuation = (max_p - min_p) / min_p * 100  # 计算波动百分比
            else:
                fluctuation = 0

            # 3. 判断波动状态变化
            current_status = self.fluctuation_status[stock_code]

            if fluctuation <= settings.UPDOWN_SPAN:  # 波动≤0.2%
                if not current_status["in_range"]:  # 首次进入范围
                    current_status["in_range"] = True
                    current_status["start_time"] = datetime.fromtimestamp(current_time).strftime(
                        "%Y-%m-%d %H:%M:%S")  # 记录开始时间
                    ##记录震荡区间
                    updown_id = add_updown(stock_code, min_p, max_p, current_time)  ##这里需要记录区间id
                    current_status["updown_id"] = updown_id  # 保存ID到状态
                    msg = f"{stock_name} 进入横盘,低位价格{min_p:.3f}"
                    self.log(msg)
                    # update_stock_updown(stock_code, updown_id)
                    # show_notification_thread("振幅提示", msg)


            else:  # 波动>0.2%
                if current_status["in_range"]:  # 刚从范围内突破
                    duration = current_time - datetime.strptime(current_status["start_time"],
                                                                "%Y-%m-%d %H:%M:%S").timestamp()
                    self.log(f"[结束] {stock_name} 脱离低波动区间，持续时间: {duration:.1f}秒")
                    update_updown(current_status["updown_id"], current_time)
                    current_status["in_range"] = False
                    current_status["start_time"] = ""
                    current_status["updown_id"] = 0  # 清除ID
                    msg = f"{stock_name} 脱离横盘"
                    # update_stock_updown(stock_code, 0)
                    # show_notification_thread("振幅提示", msg)
                    ##出波段撤买单
            return current_status["updown_id"]
    ##启动
    def start(self):
        self.log_signal(f"开启实时行情")
        self._running = True
        
        # 启动异步指标计算器
        self.async_calculator.start()
        
        # 启动数据预加载
        self.data_preloader.preload_all_data()
        
        # 启动性能监控
        self.performance_monitor.start_monitoring()

        # MACD信号预热：加载当天已有的金叉/死叉信号，避免重启后重复触发
        self._warmup_macd_signals()

        # 5分钟九转预热：为所有需要5分钟九转的股票进行预热
        try:
            _prev_verbose_5m = getattr(self.nine_turn5m, 'verbose', True)
            try:
                self.nine_turn5m.verbose = False
            except Exception:
                pass
            
            # 获取需要5分钟九转的股票列表（除了使用波动检测的股票）
            nine_turn5m_codes = [code for code in self.stock_codes 
                                  if code not in ["513130.SH", "513060.SH", "159509.SZ"]]
            
            today = datetime.now().strftime("%Y%m%d")
            
            for code in nine_turn5m_codes:
                try:
                    self.log(f"开始5分钟九转预热：{code}")
                    xtdata.download_history_data(stock_code=code, period="tick", start_time=today, end_time=today)
                    time.sleep(self.NINE_TURN_DOWNLOAD_DELAY)
                except Exception:
                    pass

                data = xtdata.get_local_data(stock_list=[code], period='tick', start_time=today, end_time=today)
                df = data.get(code) if data else None
                if df is not None and not df.empty:
                    time_col = next((c for c in df.columns if 'time' in c.lower()), None)
                    price_col = next((c for c in ['price', 'lastPrice', 'close', 'last', 'current'] if c in df.columns),
                                     None)
                    if time_col and price_col:
                        if str(df[time_col].dtype) in ('int64', 'float64'):
                            df['__time__'] = pd.to_datetime(df[time_col], unit='ms') + pd.Timedelta(hours=8)
                        else:
                            df['__time__'] = pd.to_datetime(df[time_col], format='%Y%m%d%H%M%S', errors='coerce')
                        df = df[df['__time__'].notna()].sort_values('__time__')
                        hist = pd.DataFrame({'price': df[price_col].astype(float).values}, index=df['__time__'])
                        self.nine_turn5m.warmup_from_history(code, hist)
                        
                        # 输出预热后的状态
                        down_count = self.nine_turn5m.get_down_count(code)
                        up_count = self.nine_turn5m.get_up_count(code)
                        self.log(f"5分钟九转预热完成：{code}，下跌计数={down_count}，上涨计数={up_count}")
                    else:
                        self.log(f"5分钟九转预热失败：{code} 无法识别时间或价格列")
                else:
                    self.log(f"5分钟九转预热失败或无数据：{code}")
        except Exception as e:
            self.log(f"5分钟九转预热异常: {e}")
        finally:
            try:
                self.nine_turn5m.verbose = _prev_verbose_5m
            except Exception:
                pass



        # 订阅实时行情
        for code in self.stock_codes:
            xtdata.subscribe_quote(
                code,
                period=self.period,
                count=-1,
                callback=self.quote_callback
            )
            self.log(f"已订阅 {code} 的实时行情")
        self.log(f"开始监听 {len(self.stock_codes)} 只股票的实时行情...")

        try:
            xtdata.run()  # 阻塞
        except KeyboardInterrupt:
            self.log("行情监控已手动停止")
        except Exception as e:
            self.log(f"行情监控异常: {e}")
            # 行情监控断开提醒
            show_notification_thread("行情监控", "行情监控断开")
    ##停止
    def stop(self):
        # 这里需要你根据xtdata的API实现停止订阅和run
        self._running = False
        
        # 停止异步指标计算器
        self.async_calculator.stop()
        
        # 停止性能监控
        self.performance_monitor.stop_monitoring()
    ##分钟聚合
    def _to_minute_bucket(self, ts) -> Optional[str]:
        try:
            if ts is None:
                return None
            if isinstance(ts, (int, float)):
                t = float(ts)
                if t > 1e12:
                    dt = datetime.fromtimestamp(t / 1000)
                else:
                    dt = datetime.fromtimestamp(t)
                return dt.strftime('%Y-%m-%d %H:%M')
            s = str(ts)
            if s.isdigit() and len(s) == 14:
                dt = datetime.strptime(s, '%Y%m%d%H%M%S')
                return dt.strftime('%Y-%m-%d %H:%M')
            try:
                dt = pd.to_datetime(s, errors='coerce')
                if pd.isna(dt):
                    return None
                if getattr(dt, 'tzinfo', None) is not None:
                    dt = dt.tz_convert('Asia/Shanghai').tz_localize(None)
                return dt.strftime('%Y-%m-%d %H:%M')
            except Exception:
                return None
        except Exception:
            return None
    ##上一分钟数据
    def _is_session_minute(self, minute_str: Optional[str]) -> bool:
        try:
            if not minute_str:
                return False
            dt = datetime.strptime(minute_str, '%Y-%m-%d %H:%M')
            t = dt.time()
            return (self.MORNING_START <= t <= self.MORNING_END) or (
                    self.AFTERNOON_START <= t <= self.AFTERNOON_END)
        except Exception:
            return False

    ##检查前10分钟波动率是否过低（仅下午检查）
    def _is_volatility_too_low(self, stock_code: str) -> bool:
        """检查前10分钟内的tick最高价格和最低价格的波动率是否小于0.3%（仅下午1点后检查）"""
        try:
            current_time = time.time()
            cutoff_time = current_time - (self.volatility_check_period * 60)  # 10分钟前的时间戳
            
            # 获取10分钟内的价格数据
            price_history = self.price_history_10min.get(stock_code, deque())
            if len(price_history) < 2:
                return False  # 数据不足，不跳过交易
            
            # 过滤出10分钟内的数据
            recent_prices = []
            for item in price_history:
                if item['timestamp'] >= cutoff_time:
                    recent_prices.append(item['price'])
            
            if len(recent_prices) < 2:
                return False  # 10分钟内数据不足，不跳过交易
            
            # 计算最高价和最低价
            max_price = max(recent_prices)
            min_price = min(recent_prices)
            
            if min_price <= 0:
                return False  # 价格异常，不跳过交易
            
            # 计算波动率
            volatility = (max_price - min_price) / min_price
            
            # 如果波动率小于阈值，则暂停交易
            if volatility <= self.volatility_threshold:
                stock_name = self.stock_type.get(stock_code, stock_code)
                self.log_signal(f"📊 {stock_name} 前10分钟波动率过低: {volatility:.4f} < {self.volatility_threshold:.4f} (最高:{max_price:.3f}, 最低:{min_price:.3f})")
                return True
            
            return False
            
        except Exception as e:
            self.log(f"⚠️ {stock_code} 波动率检查异常: {e}")
            return False

    ##检查成交额是否不足
    def _is_trading_amount_insufficient(self, stock_code: str) -> bool:
        """检查5分钟内跌幅超过0.5%或成交额小于5亿（成交额检查仅在下午1点之后）"""
        try:
            # 获取交易总量信息
            volume_info = get_stock_trading_volume(stock_code)
            if volume_info is None:
                # 无法获取成交额信息时，不跳过计算
                return False
            
            stock_name = self.stock_type.get(stock_code, stock_code)
            should_skip = False
            skip_reason = ""
            
            # 1. 先检查5分钟内跌幅是否超过0.5%（全天检查）
            decline_5min = volume_info.get('decline_5min')
            if decline_5min is not None:
                decline_percent = decline_5min.get('decline_percent', 0)
                if decline_percent < -2:  # 跌幅超过0.5%（负值表示下跌）
                    should_skip = True
                    skip_reason = f"5分钟跌幅{decline_percent:.2f}%"
            
            # 2. 再检查成交额是否小于5亿（仅在下午1点之后检查）
            if not should_skip:  # 如果5分钟跌幅检查未触发，才检查成交额
                total_amount_yi = volume_info.get('total_amount_yi', 0)
                ##这里上午判断1个亿下午判断5个亿
                current_time = datetime.now().time()
                if current_time >= self.AFTERNOON_START:  # 下午1点之后才检查成交额
                    if total_amount_yi < 5.0:
                        should_skip = True
                        skip_reason = f"下午成交额不足5亿({total_amount_yi:.2f}亿)"
                # else:
                #     if total_amount_yi < 1.0:
                #         should_skip = True
                #         skip_reason = f"上午成交额不足1亿({total_amount_yi:.2f}亿)"

                #
                # if total_amount_yi < 5.0:
                #     # 检查当前时间是否在下午1点之后
                #     current_time = datetime.now().time()
                #     if current_time >= self.AFTERNOON_START:  # 下午1点之后才检查成交额
                #         should_skip = True
                #         skip_reason = f"成交额不足5亿({total_amount_yi:.2f}亿)"
            
            # 如果满足跳过条件，记录日志并返回True
            if should_skip:
                return True
            
            return False
            
        except Exception as e:
            self.log(f"⚠️ {stock_code} 成交额检查异常: {e}")
            return False